#include "wrk_worker.h"
#include "include/wrk_core_t.h"
#include "wrk_logger.h"
#include <sstream>
#include "server.h"
#include "net.h"
#include <signal.h>
#include "wrk_core.h"
#include "include/zmalloc.h"
#include <sys/stat.h>
#include <fcntl.h>

#define WORKER_DIR "/tmp/dist-wrk-data/"

WrkWorker:: WrkWorker(/* args */)
{
    wrk_core_ptr = new WrkCore();
}

WrkCore *WrkWorker::wrk_core_ptr = NULL;
aeEventLoop *WrkWorker::worker_loop = NULL;

WrkWorker::~WrkWorker()
{
    delete wrk_core_ptr;
}

uint32_t WrkWorker::taskid = 0;

void WrkWorker::run(int fd)
{
    LOG_INFO("worker(pid=%d, channelid=%d) is running...", getpid(), fd);

    signal(SIGPIPE, SIG_IGN);

    signal(SIGTERM, signal_exit);

    WrkWorker::worker_loop = aeCreateEventLoop(64*1024 -1);

    aeCreateFileEvent(WrkWorker::worker_loop, fd, AE_READABLE, read_chnnel_from_master, NULL);

    // aeCreateTimeEvent(worker_loop, 5000,  monitor_stop_duration, NULL, NULL);

    aeMain(WrkWorker::worker_loop);

    aeDeleteEventLoop(WrkWorker::worker_loop);
}

void WrkWorker::signal_exit(int sig)
{
    LOG_DEBUG("worker(%d) exit(0) self", getpid());
    exit(0);
}

uint32_t WrkWorker::get_cpus_cores()
{
    return sysconf(_SC_NPROCESSORS_ONLN);
}

string WrkWorker::get_safe_json_string(const string &str)
{
    string out = string();
    for(string::const_iterator it=str.begin(); it!=str.end(); ++it) {
        switch(*it) 
        {
            case '\\':
            case '"':
                out.push_back('\\');
                break;
        }
        
        out.push_back(*it);
    }

    return out;
}

wrk_cmd_handler_t WrkWorker::cmd_handlers[] = {
    { 1, "start", WrkWorker::worker_start_task_handler},
    { 2, "stop", WrkWorker::worker_stop_task_handler},
    { 3, "stats", WrkWorker::worker_stats_task_handler},
    { 4, "upload", WrkWorker::worker_upload_file_handler},
    { 0, NULL, NULL}
};

void WrkWorker::read_file_from_client(aeEventLoop *eventLoop, int fd, void *data, int mask) 
{
    client_connection_t *c = (client_connection_t *)data;
    ssize_t n = 0;
    int ret = server::wrk_read(fd, c->read_ss, &n);

    if(ret == socket_status_t::ERROR) {
        LOG_ERROR("will aeDeleteFileEvent(eventLoop, %d, AE_READABLE|AE_WRITABLE), read error: %s", fd, strerror(errno));
        aeDeleteFileEvent(eventLoop, fd, AE_READABLE|AE_WRITABLE);
        // shutdown(fd, SHUT_RD);
        close(fd);
        delete c;
    }


    if(c->read_ss.str().length() != c->ch.length) {
        return ;
    }

    stringstream path;
    stringstream ss;
    ssize_t writen = 0;
    int file_fd = -1;
    path << WORKER_DIR << c->ch.taskid << "/";

    // 目录不存在(可读可写），则创建目录
    if (access(path.str().c_str(), R_OK|F_OK|W_OK)) {
        if(mkdir(path.str().c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH)) {
            c->write_ss.str("{\"success\":false, \"message\":\"mkdir dir "+ path.str() +" failed:"+ strerror(errno) + " \"}\n");
            goto RESP;
        }
    }
    
    path << c->ch.name;

    file_fd = open(path.str().c_str(), O_CREAT|O_RDWR, S_IRUSR|S_IWUSR);
    ftruncate(file_fd, 0);
    server::wrk_write(file_fd, c->read_ss.str().c_str(), c->ch.length, &writen);
    close(file_fd);

    ss.str("");
    if(writen != c->ch.length) {
        c->write_ss.str("{\"success\":false, \"message\":\"save file "+ path.str() +" failed\"}\n");
        CDEBUG << "readn = "<<c->ch.length << ", writen = "<<writen <<endl;
    }
    else {
        c->write_ss.str("{\"success\":true, \"message\":\"save file "+ path.str() +" success\"}\n");
    }

RESP:
    c->writen = 0;
    // 业务逻辑处理结束后开启写监听
    if(aeCreateFileEvent(WrkWorker::worker_loop, c->ch.fd, AE_WRITABLE, write_client, c) !=AE_OK ) {
        LOG_ERROR("aeCreateFileEvent(fd = %d) error: %s", c->ch.fd, strerror(errno));
        aeDeleteFileEvent(WrkWorker::worker_loop, fd, AE_WRITABLE | AE_READABLE);
        close(c->ch.fd);
        delete c;
    }
}

/**
 * @name: 
 * @test: 
 * @msg: 从客户端读取脚本，完成后进入测试。
 * @param {type} 
 * @return: 
 */
void WrkWorker::read_script_from_client(aeEventLoop *eventLoop, int fd, void *data, int mask) 
{
    client_connection_t *c = (client_connection_t *)data;
    try
    {
        ssize_t n = 0;
        int ret = server::wrk_read(fd, c->read_ss, &n);

        if(ret == socket_status_t::ERROR) {
            LOG_ERROR("will aeDeleteFileEvent(eventLoop, %d, AE_READABLE|AE_WRITABLE), read error: %s", fd, strerror(errno));
            aeDeleteFileEvent(eventLoop, fd, AE_READABLE|AE_WRITABLE);
            // shutdown(fd, SHUT_RD);
            close(fd);
            delete c;
        }


        if(c->read_ss.str().length() != c->ch.length) return ;

        stringstream path;
        path << WORKER_DIR << c->ch.taskid;
        // 目录不存在(可读可写），则创建目录
        if (access(path.str().c_str(), R_OK|F_OK|W_OK)) {
            if(mkdir(path.str().c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH)) {
                throw string("mkdir dir "+ path.str() +" failed:"+ strerror(errno));
            }
        }

        chdir(path.str().c_str());
        // wrk_core_ptr->set_stop_flag(wrk_core_state_t::START);
        WrkCore::stop = wrk_core_state_t::START;
    
        /* code */
        c->write_ss.clear();
        wrk_core_ptr->do_test(c->read_ss.str().c_str());
        c->write_ss.str("{\"success\":true, \"message\":\"start success\"}\n");
    }
    catch(const string &ss)
    {
        std::cout<< ss <<std::endl;
        WrkCore::stop = wrk_core_state_t::STOP;
        kill(server::master_pid, SIGUSR1);      
        c->write_ss.str("{\"success\":false, \"message\":\"" + get_safe_json_string(ss) + "\"}\n");
    }
    catch(...) {
        WrkCore::stop = wrk_core_state_t::STOP;
        kill(server::master_pid, SIGUSR1);      
        c->write_ss.str("{\"success\":false, \"message\":\"unkown\"}\n");
    }

    c->writen = 0;
    // 业务逻辑处理结束后开启写监听
    if(aeCreateFileEvent(WrkWorker::worker_loop, c->ch.fd, AE_WRITABLE, write_client, c) !=AE_OK ) {
        LOG_ERROR("aeCreateFileEvent(fd = %d) error: %s", c->ch.fd, strerror(errno));
        aeDeleteFileEvent(WrkWorker::worker_loop, fd, AE_WRITABLE | AE_READABLE);
        close(c->ch.fd);
        delete c;
    }
}

void WrkWorker::worker_upload_file_handler(int fd, wrk_channel_t ch, void *data)
{
    client_connection_t *client_c = (client_connection_t *)data;

    if(aeCreateFileEvent(WrkWorker::worker_loop, ch.fd, AE_READABLE, read_file_from_client, client_c) !=AE_OK ) {
        client_c->write_ss.str("{\"success\":false, \"message\":\"aeCreateFileEvent error: "+ string(strerror(errno)) +"\"}\n");
    }

}


void WrkWorker::worker_start_task_handler(int fd, wrk_channel_t ch, void *data)
{
    client_connection_t *client_c = (client_connection_t *)data;

    if(aeCreateFileEvent(WrkWorker::worker_loop, ch.fd, AE_READABLE, read_script_from_client, client_c) !=AE_OK ) {
        client_c->write_ss.str("{\"success\":false, \"message\":\"aeCreateFileEvent error: "+ string(strerror(errno)) +"\"}\n");
    }
}

void WrkWorker::worker_stats_task_handler(int fd, wrk_channel_t ch, void *data)
{
    client_connection_t *c = (client_connection_t *)data;
    LOG_DEBUG("will get stats, taskid=%d", c->ch.taskid);
    c->write_ss.str(WrkWorker::get_stats());
    c->writen = 0;

    if(WrkCore::stop==wrk_core_state_t::STOP && WrkCore::cfg.wrk_core_task_state == wrk_core_config_t::RUNNING) {
        send_release_sig_to_master();
    }

    // 业务逻辑处理结束后开启写监听
    if(aeCreateFileEvent(WrkWorker::worker_loop, c->ch.fd, AE_WRITABLE, write_client, c) !=AE_OK ) {
        LOG_ERROR("aeCreateFileEvent(fd = %d) error: %s", c->ch.fd, strerror(errno));
        aeDeleteFileEvent(WrkWorker::worker_loop, fd, AE_WRITABLE | AE_READABLE);
        close(c->ch.fd);
        delete c;
    }
}

void WrkWorker::worker_stop_task_handler(int fd, wrk_channel_t ch, void *data)
{
    client_connection_t *c = (client_connection_t *)data;
    LOG_DEBUG("command(ch.command=%d), stop task(id=%d)", ch.command, WrkWorker::taskid);
    if(WrkCore::stop==wrk_core_state_t::START) {
        WrkCore::stop = wrk_core_state_t::STOP;
        LOG_DEBUG("stop task(id=%d) success", WrkWorker::taskid);
    }

    c->write_ss.str("{\"success\":true, \"message\":\"stop success\"}\n");
    c->writen = 0;
    // 业务逻辑处理结束后开启写监听
    if(aeCreateFileEvent(WrkWorker::worker_loop, c->ch.fd, AE_WRITABLE, write_client, c) !=AE_OK ) {
        LOG_ERROR("aeCreateFileEvent(fd = %d) error: %s", c->ch.fd, strerror(errno));
        aeDeleteFileEvent(WrkWorker::worker_loop, fd, AE_WRITABLE | AE_READABLE);
        close(c->ch.fd);
        delete c;
    }

}

int WrkWorker::monitor_stop_duration(aeEventLoop *loop, long long id, void *data) 
{
    
    //master进程异常退出了，被托管给了1号进程
    if(getppid() !=server::master_pid) {
        LOG_ERROR("worker process has deattch whith master, will exit(0)");
        exit(0);
    }

    if(WrkCore::stop==wrk_core_state_t::STOP && 
        WrkCore::cfg.wrk_core_task_state == wrk_core_config_t::RUNNING && 
        (WrkRecord::time_us() - WrkCore::cfg.stop_timestamp)> 60 * 1000 * 1000 ) {
        
        LOG_WARNING("task(id = %d) has stoped, but task state is runnig, and durition more than 1 min, send release signal(SIGUSR1) to master(pid:%d)",
                    WrkWorker::taskid, server::master_pid);
        send_release_sig_to_master();
    }
    
    return 5000;
}


void WrkWorker::send_release_sig_to_master() 
{

    WrkCore::cfg.wrk_core_task_state = wrk_core_config_t::IDEL;
    pid_t ppid = getppid();
    if(ppid == server::master_pid) {
        kill(server::master_pid, SIGUSR1);
    }
    // else {
    //     LOG_ERROR("getppid()=%d, server::master_pid=%d, master process have exited, worker process should exit!", ppid, server::master_pid);
    //     exit(0);
    // }
}


void WrkWorker::write_client(aeEventLoop *eventLoop, int fd, void *data, int mask) 
{
    //关读监听
    aeDeleteFileEvent(eventLoop, fd, AE_READABLE);
    client_connection_t *c = (client_connection_t *)data;

    ssize_t writen = 0;

    int ret = server::wrk_write(fd, (char *)c->write_ss.str().c_str() + c->writen, c->write_ss.str().length() - c->writen, &writen);
    if(ret == socket_status_t::ERROR) {
        goto ERR;
    }
    c->writen += writen;
    if(c->writen == c->write_ss.str().length()) {
        aeDeleteFileEvent(eventLoop, fd, AE_WRITABLE);
        // shutdown(fd, SHUT_WR);
        LOG_DEBUG("write to client finish, close fd=%d, aeDeleteFileEvent(eventLoop,%d, AE_WRITABLE)", fd,fd);
        close(fd);
        delete c;
        c = NULL;
    }

    return;

    ERR:
        LOG_ERROR("will close fd=%d,aeDeleteFileEvent(eventLoop,%d, AE_WRITABLE) because write error: %s", fd, fd, strerror(errno));
        aeDeleteFileEvent(eventLoop, fd, AE_READABLE|AE_WRITABLE);
        close(fd);
        delete c;
        c = NULL;
}



void WrkWorker::read_chnnel_from_master(aeEventLoop *eventLoop, int fd, void *data, int mask)
{
    LOG_DEBUG("woker(pid:%d) receive from master", getpid());
    wrk_channel_t ch;
    ssize_t n;
    n = wrk_read_channel(fd, &ch, sizeof(wrk_channel_t));
   
    if(n==WRK_ERROR) {
        // socketpair正常不会关闭，此时说明父进程（master)异常退出了
        LOG_ERROR("worker will exit(0), read channel(chnnelid=%d) from master(pid:%d) error:%s", fd, getppid(), strerror(errno));
        aeDeleteFileEvent(eventLoop, fd, AE_WRITABLE | AE_READABLE);
        close(ch.fd);
        exit(0);
    }

    WrkWorker::taskid = ch.taskid;
    // wrk_core_config_t cfg = wrk_core_ptr->get_core_config();
    LOG_DEBUG("read channel(chnnelid=%d, ch.fd=%d) from master(pid:%d) success", fd, ch.fd, getppid());

    client_connection_t *client_c = new client_connection_t();
    client_c->read_ss.clear();
    client_c->read_ss.str("");
    client_c->write_ss.clear();
    client_c->write_ss.str("");
    client_c->writen = 0;
    client_c->ch = ch;

    wrk_cmd_handler_t *p = WrkWorker::cmd_handlers;
    for(; p->name!=NULL; p++) {
        if(p->cmd == ch.command) {
            LOG_DEBUG("worker aciton: %s", p->name);
            break;
        }
    }

    if(p->name ==NULL) {
        LOG_WARNING("worker unkown cmd");
        close(ch.fd);
    }

    p->handler(ch.fd, ch, client_c);

    // //业务逻辑处理结束后开启写监听
    // if(aeCreateFileEvent(WrkWorker::worker_loop, ch.fd, AE_WRITABLE, write_client, client_c) !=AE_OK ) {
    //     LOG_ERROR("aeCreateFileEvent(fd = %d) error: %s", ch.fd, strerror(errno));
    //     aeDeleteFileEvent(WrkWorker::worker_loop, fd, AE_WRITABLE | AE_READABLE);
    //     close(ch.fd);
    //     delete client_c;
    // }
}

string WrkWorker::get_stats() 
{
    return wrk_core_ptr->get_stats();
}

